Brings the working delta-kernel-rs integration over from contrib-delta-pr2
without the contrib SPI plumbing Parth flagged on apache#4339.

contrib/delta/native/:
  - jni.rs, scan.rs, engine.rs, error.rs, predicate.rs, dv_filter.rs --
    ported verbatim from contrib-delta-pr2 (only crate::proto::* import paths
    needed adjustment, handled via lib.rs re-export of the typed messages
    that now live in core's proto crate)
  - planner.rs -- Delta-specific helpers (build_delta_partitioned_files,
    parse_delta_partition_scalar with the DATE -> TIMESTAMP_NTZ widening
    fallback already inlined, ColumnMappingFilterRewriter) exposed as
    pure-DataFusion functions that core's dispatcher arm composes onto the
    standard parquet datasource path. NO ContribOperatorPlanner trait, NO
    ContribPlannerContext, NO ParquetDatasourceParams -- the contrib crate
    is now a plain library with public functions.
  - lib.rs -- module decls + a `pub mod proto` re-export of the six typed
    Delta messages from `datafusion_comet_proto::spark_operator`. No
    `#[ctor]` and no `register_contrib_planner` call.
  - Cargo.toml -- standalone (outside the native/ workspace root), no
    comet-contrib-spi dep, all delta-specific deps stay confined here.

native/core/src/execution/planner/contrib_delta_scan.rs (new):
  - `PhysicalPlanner::plan_delta_scan` -- the `OpStruct::DeltaScan` arm body
    extracted into its own file (~210 lines, mirrors `OpStruct::IcebergScan`
    in size and shape). Gated `#[cfg(feature = "contrib-delta")]`; calls
    core's `init_datasource_exec`, `prepare_object_store_with_configs`,
    `convert_spark_types_to_arrow_schema` directly + comet-contrib-delta's
    helpers for the Delta-specific pieces.

native/core/src/execution/planner.rs:
  - `OpStruct::DeltaScan` arm: 6-line dispatcher that calls into
    `self.plan_delta_scan(...)` under `#[cfg(feature = "contrib-delta")]`.

native/core/src/parquet/parquet_exec.rs:
  - New `ignore_missing_files: bool` arg on `init_datasource_exec`.
    Threaded through to `IgnoreMissingFileSource` wrapper (ported verbatim
    from PR2's native/core/src/parquet/missing_file_tolerant.rs) which
    decorates the final FileSource so its FileOpener swallows object-store
    NotFound errors as empty streams. Matches Spark's
    `spark.sql.files.ignoreMissingFiles=true` semantics. All existing call
    sites updated to pass `false`.

Build verification (both checked clean):
  cargo check  -p datafusion-comet                          # default
  cargo check  -p datafusion-comet --features contrib-delta

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

These five files port verbatim from contrib-delta-pr2 -- they touch only
Spark APIs (via reflection) and standard Scala, none of the rejected SPI
surface:

  - DeltaConf.scala               Config keys (COMET_DELTA_NATIVE_ENABLED, ...)
  - Native.scala                  JNI bridge for planDeltaScan
  - DeltaReflection.scala         Reflective access to spark-delta internals
                                  (isDeltaFileFormat, isBatchFileIndex,
                                  extractBatchAddFiles, ...)
  - RowTrackingAugmentedFileIndex Wraps a FileIndex to inject row-tracking
                                  metadata columns
  - DeltaInputFileBlockHolder     Thread-local replacement for
                                  InputFileBlockHolder on the Delta scan path

Plus the regression infrastructure (4.1.0.diff, run-test.sh,
run-regression.sh).

The remaining four files (CometDeltaNativeScan, CometDeltaNativeScanExec,
DeltaScanRuleExtension, DeltaOperatorSerdeExtension, DeltaPlanDataInjector)
each reference the rejected SPI surface (CometOperatorSerde,
CometScanRuleExtension, ContribOp envelope, PlanDataSource, PlanDataInjector).
Those need rewriting before they can compile against main -- queued as the
next commit on this branch:
  - drop the `extends CometOperatorSerde[CometScanExec]` trait bound;
    expose `convert(...)` as a static method
  - replace ContribOp envelope with the typed OpStruct::DeltaScan
  - drop the SPI extension class wrappers; integrate detection directly
    into CometScanRule.scala + CometExecRule.scala (Iceberg-style)
  - bake DeltaPlanDataInjector logic directly into CometDeltaNativeScanExec

Maven `-Pcontrib-delta` profile, scalastyle wiring, and the SPI rewrite
all land together in the follow-up commit so the contrib compiles
end-to-end against main.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ction bridge

The four SPI-touching files from contrib-delta-pr2 rewritten to drop the
rejected SPI base classes and use the typed `OpStruct::DeltaScan` proto
variant directly:

  - CometDeltaNativeScan.scala  no longer `extends CometOperatorSerde`;
    plain object with `convert(scan, builder, childOp*)` static method.
    All `ContribOp` envelope wrapping replaced with
    `builder.setDeltaScan(...)`. DeltaOperator.* imports redirected to
    core's `org.apache.comet.serde.OperatorOuterClass`.
  - CometDeltaNativeScanExec.scala  no longer `with PlanDataSource`;
    public accessors (planDataSourceKey, planDataCommonBytes,
    planDataPerPartitionBytes) stay so core's CometExecRDD can read them
    directly. `nativeOp.getContribOp.getPayload` calls collapse to the
    typed `nativeOp.getDeltaScan` accessor.
  - DeltaScanRule.scala  was `class DeltaScanRuleExtension extends
    CometScanRuleExtension`; now a plain `object DeltaScanRule` with a
    single static entry point `transformV1IfDelta(plan, session,
    scanExec, relation): Option[SparkPlan]`. The private
    `CometScanRule.isSchemaSupported` is unreachable from contrib, so
    inline the equivalent check (CometScanTypeChecker + fallback-reason
    emission).
  - The DeltaOperatorSerdeExtension + DeltaPlanDataInjector files are
    not ported -- their roles fold into the next commit's CometExecRule
    Delta serde dispatch and into CometDeltaNativeScanExec respectively.

Core wiring:
  - spark/pom.xml: new `<profile id="contrib-delta">` adds
    contrib/delta/src/main/scala/ as a compile source on comet-spark and
    pulls in `io.delta:delta-spark_2.13:4.1.0` at provided scope.
  - CometScanRule.scala: 5-line Delta detection block at the head of
    `transformV1Scan`'s HadoopFsRelation case (Iceberg-style; calls into
    `DeltaIntegration.transformV1IfDelta` which is a no-op when the
    contrib isn't bundled).
  - DeltaIntegration.scala (new): reflection bridge that resolves the
    contrib's `DeltaScanRule` + `CometDeltaNativeScan` companion objects
    by class name. Default builds get `None`; -Pcontrib-delta builds get
    a working delegate. No SPI / ServiceLoader / registry.

Build verification:
  mvn compile                  # default: still green
  mvn compile -Pcontrib-delta  # GREEN -- this is the milestone

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…tics

Spark's UnsafeRow.getUTF8String wraps bytes via UTF8String.fromAddress with
no UTF-8 validation, and cast(BinaryType -> StringType) is a zero-copy
reinterpret that leaves arbitrary bytes in a StringType column. Delta's
Z-Order uses interleave_bits(...).cast(StringType) for opaque sort keys,
which panicked Comet's strict from_utf8(...).unwrap() and cascaded into
JVM classloader errors (60+ ServiceConfigurationError tests in the
contrib-delta-pr2 regression run).

Switch to from_utf8_unchecked since the bytes flow directly into Arrow's
StringBuilder::append_value and are never introspected as a &str.

Verified on contrib-delta-pr2: OptimizeZOrderScalaSuite "interleaving"
4/4 PASS after this fix.

Pure core fix -- independent of the contrib/delta integration. Lands on
this branch because it's a prerequisite for the Delta regression to be
meaningful (without it the Z-Order panic poisons every following test).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Connects core's CometExecRule to the contrib's Delta scan serde so the
Delta-marker CometScanExec produced by CometScanRule flows through the
same `convertToComet(scan, handler)` path as Iceberg / NativeScan / etc.

  - CometDeltaNativeScan re-extends core's `CometOperatorSerde` trait
    (the trait itself is core, not part of the rejected extension SPI;
    every Comet operator handler implements it). `getSupportLevel` /
    `enabledConfig` / `convert` now properly override.
  - DeltaIntegration.scanHandler: a single reflective lookup exposes
    the contrib's companion as a `CometOperatorSerde[CometScanExec]`.
    Returns None on default builds.
  - CometExecRule.transform: new case beside the SCAN_NATIVE_DATAFUSION
    one that recognises the Delta scan marker (scanImpl ==
    "native_delta_compat") and dispatches via the handler.

Build verification:
  mvn compile                  GREEN
  mvn compile -Pcontrib-delta  GREEN

Still pending for end-to-end:
  - per-partition task-list injection (replaces PR2's DeltaPlanDataInjector
    SPI) -- baked into CometExecRDD via another small reflection hook
  - live smoke test once the dylib is rebuilt with --features contrib-delta
    and bundled into the jar

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Connects the contrib's per-partition Delta task-list serialisation into
core's existing `PlanDataInjector.injectPlanData` pipeline. Without this
the native side decodes a tasks-empty `DeltaScan` and returns `EmptyExec`
(0 rows) for every Delta scan.

  - contrib/delta/.../DeltaPlanDataInjector.scala: implements core's
    `PlanDataInjector` trait. `canInject` checks `op.hasDeltaScan` and
    rejects already-injected operators (idempotent). `inject` splices the
    partition's tasks into the operator's common-only DeltaScan envelope
    via `op.toBuilder.setDeltaScan(...)` -- pure typed-proto operations,
    no `ContribOp` envelope.
  - spark/.../operators.scala: `PlanDataInjector.injectors` Seq now
    appends the contrib injector via one reflective Class.forName lookup.
    Default builds get None (no contrib classes on classpath) so the
    list is unchanged; -Pcontrib-delta builds get the Delta injector.

Build verification:
  mvn compile -Pcontrib-delta  GREEN

End-to-end Scala+Maven integration is now complete. Remaining work:
  - rebuild native dylib with `--features contrib-delta` and bundle
    into comet-spark.jar
  - run an isolated test (e.g. OptimizeZOrderScalaSuite "interleaving")
    to confirm the end-to-end path works

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Wrap Class.forName calls in `// scalastyle:off classforname`, change
Option[Class[_]] to Option[Class[AnyRef]] to avoid existential type
warnings, reword the doc comment so the verbatim string Class.forName
doesn't trip scalastyle's source-pattern check.

  mvn scalastyle:check -Pcontrib-delta  GREEN

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…th file path

CometExecIterator was wrapping native Parquet failures (e.g. corrupt-footer
errors from kernel-rs reading a broken Delta checkpoint) in `_LEGACY_ERROR_TEMP_2254`,
whose message is literally "Data read failed." -- no file path, no useful context.

That broke tests that mirror Spark/Delta's standard parquet-failure shape, e.g.
SnapshotManagementSuite "should not recover when the current checkpoint is broken"
which asserts the resulting SparkException's message contains both the file path
and "Encountered error while reading file" -- the format
`QueryExecutionErrors.cannotReadFilesError` produces.

Switch the wrapping to `cannotReadFilesError(cause, filePath)` via a new helper
on ShimSparkErrorConverter (which lives in the spark package and can reach the
private InputFileBlockHolder / QueryExecutionErrors). File path is read from
InputFileBlockHolder, with an empty-string fallback when the thread-local
isn't set; the static phrasing still satisfies the test assertion.

Pure core fix -- benefits every native parquet read, not just Delta.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

DeltaTable.forPath(spark, path, fsOptions) with a Hadoop custom-fs scheme
(e.g. fake://) was being claimed by CometScanRule for V1 parquet scans on
the _delta_log/checkpoint.parquet files Delta reads internally. The native
side then crashed at executePlan with `Generic URL error: Unable to
recognise URL "fake:///..."` since object_store doesn't know the custom
scheme.

Add a scheme allowlist check (same set already used in the Iceberg branch
and the contrib Delta path) at the top of the HadoopFsRelation arm; decline
via withInfo when any rootPaths scheme is outside the allowlist so Spark's
Hadoop-FS-aware reader handles the scan.

Fixes DeltaTableSuite "dropFeatureSupport - with filesystem options" and is
also a baseline fix (the same crash reproduces on main per
full-20260415-222735.log).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Each `plan_delta_scan` JNI call was creating a fresh `DefaultEngine`. Kernel's
`DefaultEngine<TokioBackgroundExecutor>` spawns one std::thread per executor that
hosts a current_thread tokio runtime, and that runtime's blocking pool (used by
kernel for parquet metadata IO and object_store reads) keeps `spawn_blocking`
worker threads alive for ~10s after each task. Under regression load (hundreds
of Delta scans/minute, each spawning a handful of blocking IO tasks) this
accumulates OS threads faster than tokio reaps them, eventually hitting the
per-process `ulimit -u` (~1300 on macOS) — visible in the log as
`pthread_create EAGAIN` aborts of GenerateIdentityValuesSuite and
MergeIntoUnlimitedMergeClausesScalaSuite ~2 hours into the run.

Replace the per-call `create_engine` with `get_or_create_engine` that returns
an `Arc<DeltaEngine>` from a static cache keyed by `(scheme, authority,
DeltaStorageConfig)`. Engines are constructed lazily on first miss per key and
reused for the lifetime of the JVM, bounding live OS threads by table-storage
diversity rather than by request count. The standalone `create_engine` is kept
(behind `#[allow(dead_code)]`) for tests that want a fresh engine.

`scan.rs` updated to deref `Arc<DeltaEngine>` to `&dyn Engine` at each kernel
call (`builder.build`, `scan.scan_metadata`, `dv.get_row_indexes`).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

DataFusion's `make_array_inner` asserts strict element-type equality (down to
nested field nullability) via `MutableArrayData::with_capacities`. Spark's
`CreateArray` is more permissive: when the analyzer doesn't insert coercion
casts, children can share the same surface struct type but disagree on a
nested field's nullability. Delta's CDF write path builds
`array(struct(id, b, _change_type=lit("delete")), struct(id, b, _change_type=col))`
manually -- one arm's `_change_type` is `Utf8` non-nullable (from a literal),
another is `Utf8` nullable -- and Comet's native serde happily emitted a
`make_array` call. Native execution then panicked:

  assertion `left == right` failed: Arrays with inconsistent types passed to
  MutableArrayData
   left: Struct([..., Field { name: "_change_type", data_type: Utf8 }])
  right: Struct([..., Field { name: "_change_type", data_type: Utf8, nullable: true }])

Decline in `CometCreateArray` when `children.map(_.dataType).distinct.size > 1`
so the JVM evaluator (which doesn't have this strictness) handles it. Fixes 4
`DescribeDeltaHistorySuite "replaceWhere on data column ... enableCDF=true"`
failures.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

… kind

Two perf-sweep items from apache#135:

apache#7 parse_delta_partition_scalar TZ parse-once. The per-row
chrono_tz::Tz::from_str (or fixed-offset parse) was happening inside
parse_delta_partition_scalar for every TIMESTAMP partition value, but the
session TZ string doesn't change within a scan. Introduce SessionTimezone
enum (Tz | Offset | Invalid), parse once in build_delta_partitioned_files,
pass the parsed value through. parse_delta_partition_scalar's signature gains
&SessionTimezone and keeps session_tz: &str only for the error message.

apache#2 PlanDataInjector lookup by op kind. injectPlanData was running
`for (injector <- injectors if injector.canInject(op))` against every
operator in the tree; for a 50-op plan with 3 injectors that's 150
canInject calls just to find no match on most ops. Add `opStructCase` to the
PlanDataInjector trait, build a Map[OpStructCase, PlanDataInjector] once at
object init, and look up by op.getOpStructCase before any canInject call.
Iceberg/NativeScan/Delta injectors set their own opStructCase.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…tion

Perf-sweep #1 from apache#135. `DeltaIntegration.transformV1IfDelta` is invoked for
every V1 scan in every plan (the bridge is called unconditionally by
CometScanRule before the contrib's own Delta-format check). On
-Pcontrib-delta builds each call was doing `getField MODULE$` +
`getMethod("transformV1IfDelta", ...)` + 4-arg Method.invoke -- a reflection
round-trip per scan.

Cache the resolved (module, method) binding once per JVM as
`transformV1IfDeltaBinding: Option[(AnyRef, Method)]`, single OnceLock-style
volatile. Steady-state per-scan cost drops to one volatile read + one
Method.invoke.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Perf-sweep apache#5 from apache#135. `isSchemaCometCompatible` was allocating a fresh
CometScanTypeChecker(CometDeltaNativeScan.ScanImpl) on every scan. The
checker is stateless w.r.t. its scanImpl tag and is safe to share. Promote
it to a private val on DeltaScanRule; the per-scan fallback-reasons
ListBuffer remains per-call (it's the only mutable input).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…artitioned_files

Perf-audit apache#137 finding #1. The inner `partition_schema.fields()` loop was
calling `.iter().find()` on `task.partition_values` for every field --
O(width × values) per task. Pre-build a per-task HashMap<&str, &str> once,
then O(1) gets. The map is reused across tasks via clear() so the allocation
amortises across all DeltaScanTasks in the scan.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

SnapshotManagementSuite "should not recover when the current checkpoint is
broken..." asserts the wrapped FAILED_READ_FILE.NO_HINT SparkException message
contains the file path (e.g. "0001.checkpoint"). de9e0d3 got the error class
right but left the path empty because:

  1. Comet's native scan path does NOT go through Spark's FileScanRDD, so the
     standard InputFileBlockHolder thread-local is never populated.
  2. ShimSparkErrorConverter.wrapNativeParquetError was reading from
     InputFileBlockHolder, getting null, and passing "" to
     cannotReadFilesError -- producing "Encountered error while reading file . "
     (with the empty path), which the test rejected.

Plumb per-partition file paths from CometNativeScanExec (where they're known
at planning time) -> CometExecRDD -> CometExecPartition -> CometExecIterator
-> wrapNativeParquetError. CometNativeExec.doExecuteColumnar (the actual call
site that constructs the iterator for query trees with a scan) collects file
paths from any CometNativeScanExec leaves and passes them through the same
CometExecRDD parameter.

Verified with a /tmp/cometdiag.log file sentinel that the existing logWarning
diags were being silently dropped by the test's `quietly { ... }` block,
which is why my earlier "the wrap isn't being reached" conclusion was wrong.

Test results after fix: SnapshotManagementSuite checkpoint-broken 2/2 PASS
(was 0/2 with empty path). The other 3 fix clusters
(de9e0d3+effe5f76+56c2b011) continue to pass: replaceWhere CDF 8/8,
dropFeatureSupport 1/1.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…+ safeguards

Five fixes from the comprehensive code review of contrib-delta-direct:

1. Implement the missing InputFileBlockHolder hook in CometExecRDD.compute.
   Several docs referenced `CometExecRDD.setInputFileForDeltaScan` but no such
   method existed and nothing called `DeltaInputFileBlockHolder.set`, leaving
   Delta's UPDATE/DELETE/MERGE flows (which use `input_file_name()` to find
   touched files) silently looking at an empty path. Now set the thread-local
   to the partition's first file (one-per-partition is enforced by
   DeltaScanRule when input_file_name() is referenced), unset on task
   completion. Stale doc references updated to point at the real call site.

2. DV filter ordering safeguards. DeltaDvFilterExec's `current_row_offset`
   tracking assumes physical row ordering from the parquet scan. Override
   `maintains_input_order() = [true]` and
   `benefits_from_input_partitioning() = [false]` so any future optimizer
   that wants to insert a RepartitionExec / SortPreservingMergeExec is
   forced to bail rather than silently re-order rows.

3. Tighten IgnoreMissingFileSource's `is_not_found` Display fallback. The
   prior `msg.contains("not found")` would match unrelated parquet messages
   like "row group statistics not found" or "page index not found" and
   silently swallow them as missing-file (returning empty results instead
   of failing). Restrict to recognised NotFound prefixes from object_store /
   S3 / FS error formats.

4. Multi-line regex for native parquet errors in CometExecIterator. Native
   parquet errors with embedded newlines (e.g. footer hex dumps) would slip
   past the single-line `^Parquet error: .*$` and surface as bare
   CometNativeException. Add `(?s)` so `.` spans newlines.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The post-review fixes added/modified scaladoc that broke spotless line-length
rules. Apply spotless:apply across the three touched files. Verified with
test-compile.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…log replay

Closes the P1 credential-asymmetry gap carried from apache#3932 (commit 461fa4f).
Previously the kernel-rs log-replay path's DeltaStorageConfig only honored
explicit static keys (`fs.s3a.access.key` / `fs.s3a.secret.key` /
`fs.s3a.session.token`) set in core-site.xml. Users running under
SimpleAWSCredentialsProvider / TemporaryAWSCredentialsProvider /
AssumedRoleCredentialProvider / IAMInstanceCredentialsProvider would see
data-file reads authenticate (those go through Comet's existing native
`build_credential_provider`) but log replay fail.

Resolution happens Scala-side via reflection against
`org.apache.hadoop.fs.s3a.S3AUtils.createAWSCredentialProviderList` -- the
same Hadoop credential machinery Spark uses everywhere else. The resolved
(access_key, secret_key, session_token) tuple is stuffed into the
`storageOptions` map under the standard Hadoop keys before the JNI call.
Reflective because hadoop-aws is an optional dep; absence falls through to
static-only behavior (any user without S3 stays unaffected).

Architecture note: an in-crate cherry-pick of 461fa4f wasn't viable here
because the JNI lives in `contrib/delta/native/` -- a standalone Cargo
crate that deliberately doesn't depend on core (to keep the arrow-57 /
arrow-58 split clean). The Scala-side approach has the same correctness
properties and avoids the crate boundary entirely.

Method handles cached via @volatile Option[Option[Binding]] -- the augment
path runs on every Delta scan; resolving the Class + getMethod chain on
each call would be a per-scan reflection round-trip just to find the same
handles every time.

SNAPSHOT resolution: log replay completes in seconds, well within any
reasonable credential TTL. Long-running data reads continue to use Comet's
refresh-capable native credential provider.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ad-bearing

Expand the comment on the CM-name + checkLatestSchemaOnRead=false guard to
explain the specific failure mode (column_mappings from one snapshot vs.
parquet physical names from another after a concurrent ALTER TABLE). The
guard is conservative but necessary; a future reader of the code shouldn't
mistake it for laziness.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…uet field IDs

Implements apache#142. Previously declined at DeltaScanRule.scala:271 because the
contrib's native path matched parquet columns by name and CM-id mode demands
ID-based matching. Comet core's `schema_adapter.rs` already supports field-ID
matching via `use_field_id` + `ignore_missing_field_id` flags; this PR wires
the Delta contrib through that machinery.

Five mechanical changes:

  1. Add `parquet.field.id` (Spark's standard StructField metadata key for
     parquet field IDs) and `delta.columnMapping.id` (Delta's CM-id storage
     key) as named constants in DeltaReflection.

  2. Add `use_field_id` bool to DeltaScanCommon proto (field 17).

  3. CometDeltaNativeScan.translateDeltaFieldIdToParquet walks the schema
     tree recursively (StructType -> nested fields, ArrayType -> element,
     MapType -> key/value) copying `delta.columnMapping.id` to
     `parquet.field.id` on every StructField. Spark's `ParquetUtils.hasFieldId`
     -- which schema2Proto and serializeDataType's StructType arm read --
     looks at `parquet.field.id`, so this is what makes the field IDs actually
     reach the proto.

  4. In `convert()`, detect CM-id mode from snapshot metadata and apply the
     translator to data_schema / required_schema / partition_schema before
     calling `schema2Proto`. Set `commonBuilder.setUseFieldId(true)` so the
     native dispatcher passes `use_field_id=true` to `init_datasource_exec`.

  5. native/core/src/execution/planner/contrib_delta_scan.rs uses
     `common.use_field_id` from the proto instead of the hardcoded `false`.

The recursive translator handles nested struct / array / map field IDs --
the "complex sub-types" gotcha from earlier CM-name work.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…n-blocking

Expand inline comments on the three remaining DeltaScanRule fallback gates
(TahoeLogFileIndexWithCloudFetch, __delta_internal_* synthetic columns,
CometScanTypeChecker decline) to document why they're correctness-correct as
fallback-only paths and to capture the implementation sketches for any future
native-perf work.

No behavioral change. Each gate was verified in the recent regression to
either never fire (cloud-fetch -- OSS Delta doesn't have the class) or fire
on a path Spark's reader handles correctly without test failures (synthetic
columns, schema type decline).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…pache#144)

Native ExecutionPlan wrapper that appends Delta's `__delta_internal_row_index`
(UInt64) and `__delta_internal_is_row_deleted` (Int32) columns to scan output
batches. Replaces the decline for these synthetic columns where the surrounding
plan asks for them (UPDATE/DELETE/MERGE flows).

- `synthetic_columns.rs`: new module with DeltaSyntheticColumnsExec. Same
  physical-order invariant as DeltaDvFilterExec (one file per partition;
  parquet emits in file row order). Appends columns via a single sweep over
  the DV-sorted indexes alongside the batch's row range.
- proto: add `emit_row_index` (18) and `emit_is_row_deleted` (19) flags on
  DeltaScanCommon.
- contrib_delta_scan.rs: wire three mutually-exclusive wrap modes -- synthetic
  exec, DV filter exec, or passthrough.

NOT YET WIRED Scala-side: when scan.requiredSchema contains these synthetic
column names, CometDeltaNativeScan still needs to (a) strip them from the
proto schemas (so the native parquet reader doesn't try to read them) and
(b) set the proto emit flags. Until that lands the existing decline gate at
DeltaScanRule.scala:331-342 stays active.

Native module compiles clean. Full linker validation deferred -- disk-space
pressure from concurrent regression run blocked the full link cycle.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…olumns

Completes apache#144. CometDeltaNativeScan.convert now:
  - Detects __delta_internal_row_index / __delta_internal_is_row_deleted in
    scan.requiredSchema
  - Verifies they form a contiguous SUFFIX of required_schema (so wrapped
    DeltaSyntheticColumnsExec's appended-at-end output matches Spark's
    expected layout); declines otherwise
  - Strips them from the proto required_schema and data_schema so the parquet
    reader doesn't look for columns that aren't on disk
  - Filters them out of projection_vector (their -1 sentinel would have been
    out-of-bounds for native usize)
  - Sets the proto emit_row_index / emit_is_row_deleted flags so the
    dispatcher wraps the parquet scan in DeltaSyntheticColumnsExec to append
    them back

DeltaScanRule: removed the decline gate at scanWithMappedSchema. Removed the
belt-and-suspenders guard in CometDeltaNativeScan now that the convert path
handles synthetics rather than falling back.

Combined with the native exec from 2cb9188, this lets UPDATE/DELETE/MERGE
flows that materialise the DV deletion flag stay on the native path instead
of falling back to Spark's Delta reader.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The decline at DeltaScanRule for "$ScanImpl does not support Parquet field ID
matching" was a separate gate from CM-id mode, fired when the user explicitly
set spark.sql.parquet.fieldId.read.enabled=true AND scan.requiredSchema
carried Spark's standard `parquet.field.id` metadata (non-Delta-id path that
nevertheless wants field-ID matching).

The same native machinery wired for CM-id (apache#142, commit 7ace165) handles
this case unchanged -- `serializeDataType`'s StructType arm reads
`ParquetUtils.hasFieldId` for nested types and `schema2Proto` does the same
for top-level. The only thing needed was setting `use_field_id=true` on the
proto.

CometDeltaNativeScan.convert now sets `useFieldIdActive` from EITHER CM-id
mode OR (Spark's PARQUET_FIELD_ID_READ_ENABLED + hasFieldIds). Gate removed
from DeltaScanRule.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…commit_version)

Unblocks the second gate in DeltaScanRule.applyRowTrackingRewrite, which used
to decline native execution when a row-tracking-enabled table HAD no
materialised column names (rowIdPhysical / rowVerPhysical both empty,
meaning Delta expects synthesis from baseRowId + physical row index).

End-to-end wiring:

  - scan.rs: extract baseRowId / defaultRowCommitVersion per scan file from
    each ScanMetadata batch's underlying RecordBatch
    (`fileConstantValues.baseRowId` / `defaultRowCommitVersion` -- not
    exposed by kernel's `ScanFile`). Uses an `RawEntryAcc` context struct
    because `visit_scan_files` requires `fn` (not `FnMut`), so the per-batch
    row-tracking lookup vec lives in the context.
  - jni.rs: thread the extracted values into DeltaScanTask proto fields 6/7
    (already present, previously hard-None'd).
  - proto: add `emit_row_id` (20) and `emit_row_commit_version` (21) flags
    on DeltaScanCommon.
  - synthetic_columns.rs: extend DeltaSyntheticColumnsExec to emit the two
    new columns (row_id = baseRowId + physical_row_index per file,
    row_commit_version = defaultRowCommitVersion constant per file). Nullable
    Int64 columns; null-valued when the file has no row tracking.
  - contrib_delta_scan.rs: force per-file FileGroups when emit_row_id /
    emit_row_commit_version is on (the per-partition row offset counter
    doesn't reset across files within a FileGroup, so baseRowId arithmetic
    requires 1:1 file-to-partition mapping just like the DV case).
  - CometDeltaNativeScan: detect row_id / row_commit_version in
    scan.requiredSchema, add to synthetic-column suffix check + strip from
    proto schemas + projection_vector, set emit flags.
  - DeltaScanRule.applyRowTrackingRewrite: stop declining the no-materialised
    case; return None (no rewrite needed) so nativeDeltaScan proceeds and
    CometDeltaNativeScan.convert sets the synthesis path.

Also unblocks the related field-id-matching gate when
spark.sql.parquet.fieldId.read.enabled is true (commit ee9f9e4) -- the
same use_field_id machinery handles both CM-id and non-CM-id paths.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

With native synthesis of `__delta_internal_is_row_deleted` wired in apache#144,
the `outputHasIsRowDeleted` branch of `scanBelowFallsBackForDvs` no longer
needs to force a decline. CometDeltaNativeScan.convert detects the column
in scan.requiredSchema and routes through DeltaSyntheticColumnsExec to
append it -- the surrounding Delta projection that filters on the column
runs against the synthesised output without falling back to Spark.

Only `batchFallback` (TahoeBatchFileIndex with DV-bearing AddFiles) still
forces decline because our native path can't extract DV info from
pre-materialised batch indexes -- separate issue.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…audit core

Move the 400-line `OpStruct::DeltaScan` planner body out of
`native/core/src/execution/planner/contrib_delta_scan.rs` and into
`contrib/delta/native/src/core_glue.rs`, co-located with the rest of the
Delta integration. The file is still compiled as part of core via a
`#[path]` attribute on the `mod` declaration -- a true cross-crate `impl
PhysicalPlanner` is forbidden by Rust, and a `contrib -> core` cargo
dependency would create a cycle with core's optional `contrib-delta`
dep, so `#[path]` is the available tool that lets the file's HOME be
with Delta while its COMPILATION unit stays in core. Build gate
(`cfg(feature = "contrib-delta")`) is preserved exactly.

Audit complete (apache#170). After the move, the only Delta references left
in `native/core/src/` are:

  - `planner.rs:33-35`  -- the `#[path]` `mod contrib_delta_scan;`
    declaration (feature-gated).
  - `planner.rs:1512-1527` -- the `OpStruct::DeltaScan` dispatcher arm
    (both halves feature-gated; default build returns a clear error).
  - `jni_api.rs:op_name`  -- exhaustive-match arm returning the string
    "DeltaScan" for tracing event names. No contrib logic; documented.
  - `planner/operator_registry.rs:to_operator_type` -- exhaustive-match
    arm returning `None`. No contrib logic; documented.

`OpStruct` is a proto-generated enum (in `datafusion-comet-proto`), so
Rust requires exhaustive matches everywhere it appears -- the un-gated
arms in jni_api.rs / operator_registry.rs are structural, not behavioural,
and keeping them un-gated is what lets default builds emit a clear
"Received a DeltaScan operator but core was built without the
`contrib-delta` Cargo feature" error message that can identify the
operator by name.

Default build (no `--features contrib-delta`) and contrib-enabled build
both compile clean. Contrib Scala suites still 49/49.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Adds `dev/verify-contrib-delta-gate.sh`, an enforceable CI-ready check
that asserts the `contrib-delta` build gate keeps Delta surface out of
default builds. Six independent checks across three layers:

  Cargo:
    1. `cargo tree -p datafusion-comet --no-default-features` has zero
       `comet-contrib-delta` / `delta_kernel` entries.
    2. `cargo tree -p datafusion-comet --features contrib-delta` DOES
       pull both -- catches a future regression that accidentally
       turns the contrib off.

  Maven:
    3. `mvn -Pspark-4.1 ... dependency:list` has zero `io.delta:*`
       dependencies.
    4. `mvn -Pspark-4.1,contrib-delta ... dependency:list` correctly
       pulls `io.delta:delta-spark`.
    5. `mvn -Pspark-4.1 ... test-compile` produces no
       `org/apache/comet/contrib/**.class` and no
       `CometDeltaNativeScan*` / `DeltaScanRule*` / `DeltaReflection*`
       classes (only the always-present `DeltaIntegration` reflection
       bridge in core).

  Native artifact:
    6. Default `libcomet.dylib` is meaningfully smaller than the
       contrib-enabled build (~57 MB delta on this machine) AND has
       zero `comet_contrib_delta` / `delta_kernel` / `DeltaDvFilter` /
       `DeltaSynthetic` external symbols.

Current run on this branch: all 6 checks PASS. Wire this into CI to
catch leaks at PR time instead of post-merge.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…tion

Adds contrib/delta/docs/07-spark35-feasibility.md -- a complexity
estimate for back-porting the contrib to Spark 3.5 + Delta 3.3, with
file-level change scope, per-Spark-API gap analysis, expected coverage
regression, and a tiered recommendation.

TL;DR:
  - Minimal viable: 2-3 dev-days. Spark-3.5 compile + most coverage,
    row-tracking unmaterialised reads fall back to vanilla (Spark 4 added
    `_metadata.row_index` / `_metadata.row_id`; Spark 3.5 lacks them).
  - Production-equivalent: 1-2 dev-weeks. Includes regression-diff port
    and CI matrix.
  - Full multi-version (3.4 + 3.5 + 4.x): 3-4 dev-weeks. Decaying ROI.

Native side has zero changes -- delta-kernel-rs 0.19 handles both
Delta 3.x and 4.x log formats transparently. All version drift is in
the JVM-side reflective accessors (deltaLog.update arity,
snapshot.filesForScan arity) and one Maven property
(<delta.version> per Spark profile).

Recommendation: defer to a follow-up PR gated on user demand.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Achieves full parity: 49/49 contrib Scala tests pass under BOTH
`spark-4.1 + Delta 4.1.0` AND `spark-3.5 + Delta 3.3.2`. The feasibility
doc I added previously was too pessimistic about gaps -- it turns out
Delta's own strategy (in BOTH Delta 3.x and 4.x) expands `_metadata.row_id`
into `coalesce(_row-id-col-<uuid>, base_row_id + _tmp_metadata_row_index)`,
and all those synthetics are things we already handle natively. No
Spark-4-only API was actually load-bearing for the contrib.

Total surface area of the change:

  1. spark/pom.xml -- move `<delta.version>` from the contrib-delta
     profile into each Spark profile (spark-4.1 -> 4.1.0, spark-3.5 ->
     3.3.2). When `-Pcontrib-delta` is layered onto a Spark profile, the
     matching Delta version is picked up automatically.

  2. spark/src/main/spark-3.5/.../ShimSparkErrorConverter.scala -- add
     `wrapNativeParquetError` mirroring the spark-4.x shim of the same
     name. `QueryExecutionErrors.cannotReadFilesError(Throwable, String)`
     has the same signature in Spark 3.5 so the implementation is identical.
     (This was a pre-existing Comet-core gap that any branch building under
     spark-3.5 + this branch's per-task file-path threading would hit.)

  3. CometDeltaTestBase.scala -- use `SparkSession.builder()` instead of
     `org.apache.spark.sql.classic.SparkSession.builder()`. The `classic`
     subpackage is a Spark 4 addition; the un-qualified path works on both
     and resolves to the same classic builder on Spark 4.

  4. dev/verify-contrib-delta-gate.sh -- extend to verify the per-Spark
     Delta version pinning: spark-3.5 + contrib-delta must pull
     `delta-spark:3.x`, spark-4.1 + contrib-delta must pull
     `delta-spark:4.x`. Catches a future bug where someone hardcodes
     the wrong Delta version.

  5. Add ASF headers to the new contrib-delta files (doc + script) to
     satisfy rat-plugin. Delete the working PR-body draft from disk.

Native side: zero changes. delta-kernel-rs 0.19 reads both Delta 3.x
and 4.x log formats, so the same libcomet works under either Spark
version.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Now that Spark 3.5 + Delta 3.3.2 support is in (49/49 contrib tests
green on both Spark profiles), rewrite the doc as a status + post-mortem
instead of a prediction. The original feasibility prediction is preserved
verbatim at the end as a record of where the analysis was wrong:
predicted "minimal viable: 2-3 dev-days", actual cost was one ~2-hour
session.

The load-bearing wrong-assumption: that the contrib leaned on Spark 4's
`_metadata.row_id` / `_metadata.row_index` virtual columns. It does not.
Delta's own `GenerateRowIDs` strategy (present in both Delta 3.x and 4.x)
expands `_metadata.row_id` into
`coalesce(_row-id-col-<uuid>, base_row_id + _tmp_metadata_row_index)`
before the plan reaches us, and all three of those synthetics are
handled by our `DeltaSyntheticColumnsExec`. Spark 4's new `_metadata`
virtual columns are irrelevant to our path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Adds .github/workflows/delta_contrib_test.yml -- three independent jobs:

  build-native (1x):
    Builds libcomet.so once with `cargo build --profile ci --features
    contrib-delta`. Cached on Cargo.lock + native sources hash. Uploaded
    as an artifact for downstream matrix cells.

  delta-contrib-scala (matrix, 2x):
    Cells: (Spark 3.5.8 + Delta 3.3.2) and (Spark 4.1.1 + Delta 4.1.0).
    Downloads the prebuilt native lib, then runs all 4 contrib Scala
    suites: CometDeltaFeaturesSuite, CometDeltaNativeSuite,
    CometDeltaColumnMappingSuite, CometDeltaCoverageSuite (49 tests each
    cell). Uploads surefire-reports on failure so PR diagnosis is local.

  delta-build-gate (1x, parallel):
    Cheap independent job. Runs dev/verify-contrib-delta-gate.sh which
    asserts default cargo + mvn + dylib carry zero Delta surface, AND
    that `-Pcontrib-delta` with each Spark profile pulls the matching
    Delta version (delta-spark:3.x for spark-3.5, delta-spark:4.x for
    spark-4.1). Catches accidental leak-into-core regressions at PR time.

Structure mirrors iceberg_spark_test.yml (paths-ignore on doc-only
changes; concurrency cancel-in-progress; RUSTFLAGS=-fuse-ld=bfd to keep
GNU ld) and supersedes PR apache#3932's delta_spark_test.yml +
delta_regression_test.yml (which targeted the older non-contrib design).
Validated locally with actionlint 1.7.12.

The Delta own-test-suite regression (port of PR apache#3932's
delta_regression_test.yml) is intentionally NOT in this commit -- it
requires the dev/diffs/delta/3.3.2.diff + 4.1.0.diff infrastructure
which is a separate body of work. Tracked under a follow-up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…e hardening)

Addresses the 8 findings from the independent code review (see PR apache#4366
comments). 49/49 contrib tests still pass on BOTH Spark 4.1 + Delta 4.1.0
and Spark 3.5 + Delta 3.3.2 after these changes.

Critical:

  1. native/shuffle/src/spark_unsafe/unsafe_object.rs: replace
     `from_utf8_unchecked` with `from_utf8_lossy` returning `Cow<'_, str>`.
     The previous version constructed a `&str` from arbitrary bytes
     (Spark's binary-cast-to-string case, e.g. Delta's Z-Order
     `interleave_bits(...).cast(StringType)`) -- the Rust reference defines
     that as UB even when the bytes only get copied downstream, because
     downstream Arrow ops internally use `str::from_utf8_unchecked` on the
     StringArray buffer and would propagate the UB. `from_utf8_lossy` is
     well-defined: zero-cost borrow for valid UTF-8, allocates a String
     with U+FFFD replacements for invalid bytes (only fires on the
     binary-cast case, which Spark never displays as text anyway). All
     call sites pass to `StringBuilder::append_value` which takes
     `AsRef<str>`; `Cow<str>`'s `AsRef<str>` impl makes them work
     transparently. No call-site changes.

  2. DeltaIntegration.scala: narrow the `case _: Exception => None`
     swallow in `transformV1IfDelta` to ONLY catch true reflection
     binding failures (`NoSuchMethodException`/`NoSuchFieldException`/
     `IllegalAccessException`) and invocation errors
     (`IllegalAccessException`/`IllegalArgumentException`). An
     `InvocationTargetException` -- the contrib's transform actually
     threw -- now log-warnings and declines instead of silently falling
     back to vanilla. Without this, kernel-rs IO errors, CCE on a Delta
     version bump, NPE in the CM-id translator etc. would silently
     decline and the user would never know. Same narrowing applied to
     `scanHandler` and `DeltaPlanDataInjector` lookup (operators.scala).

Should-fix:

  3. CometExecRDD.compute: don't set InputFileBlockHolder when a
     partition has multiple files. Previous code took
     `partition.filePaths.head` always, which would silently report the
     first file's path for every row when a contrib accidentally batched
     multiple files in one partition. (Tried `require(length == 1)`
     first; that's too strict because partitioned reads legitimately have
     multi-file partitions but don't query `input_file_name()`. Skipping
     the hook on multi-file partitions preserves correctness for
     `input_file_name()` callers -- which MUST one-task-per-partition
     anyway -- without false-positive failing legitimate partitioned
     reads.)

  4. engine.rs: LRU-bound the engine cache at MAX_CACHE_ENTRIES=32. The
     cache key included `DeltaStorageConfig` which contains
     `aws_session_token`; long-running drivers with rotating STS/IRSA
     credentials would grow one entry per rotation and LEAK one
     `TokioBackgroundExecutor` thread per stale entry. With LRU eviction,
     `Arc<DeltaEngine>` drops on eviction, `DefaultEngine` drops its
     `TokioBackgroundExecutor`, the OS thread joins, thread count
     stabilizes. Test `get_or_create_engine_evicts_lru_when_full`
     verifies the bound + eviction order.

Nits:

  5. planner.rs: error message for the "DeltaScan in default build" case
     now mentions BOTH `-Pcontrib-delta` (Maven) and `--features
     contrib-delta` (Cargo) -- previously mentioned only the Cargo flag.

  6. dev/verify-contrib-delta-gate.sh: also assert the contrib-enabled
     libcomet has >0 Delta-related external symbols. Without this, a
     future Rust toolchain change that mangles symbol names differently
     would silently turn the default-build symbol check into a no-op
     while still passing -- the gate would lie about being enforced.
     Asserting both "default has 0" AND "contrib has >0" catches grep
     pattern drift.

Build infrastructure:

  7. pom.xml + spark/pom.xml: move `<delta.version>` default to the
     parent POM's top-level properties. Per-Spark-profile `delta.version`
     overrides cleanly (spark-3.5 -> 3.3.2, spark-4.1 -> 4.1.0), and
     spotless-style invocations without a Spark profile still resolve
     the property. The previous arrangement (default in `contrib-delta`
     profile) had Spark-profile overrides silently lose to the
     contrib-delta default because of POM profile-document-order property
     precedence.

  8. Make `PlanDataInjector` and `DeltaIntegration` extend
     `org.apache.spark.internal.Logging` so the new `logWarning` calls
     compile.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

… version)

Pin Delta 4.0.0 to spark-4.0 profile -- Delta 4.1 trips NoSuchMethodError
on ParserInterface.$init$ at runtime against Spark 4.0 internals.
spark-4.1 already pins Delta 4.1.0; spark-3.5 already pins Delta 3.3.2.
Native side: zero changes (kernel reads both log formats).

Extended dev/verify-contrib-delta-gate.sh to also assert
`-Pcontrib-delta + spark-4.0` pulls delta-spark:4.0.x specifically
(catches a future regression that lets 4.1 leak in).

Extended .github/workflows/delta_contrib_test.yml matrix to include
spark-4.0 + delta-4.0.0 as a third cell alongside the existing
spark-3.5 + delta-3.3.2 and spark-4.1 + delta-4.1.0.

Test status: 49/49 contrib Scala tests pass on all THREE Spark+Delta
combinations now. Build-gate check passes for all three.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…) rebuild

CometDeltaNativeScanExec.convertBlock() reconstructed the case-class but
omitted the `oneTaskPerPartition` constructor arg, so the case-class
default (false) silently overrode any value the planner had set. Comet
calls convertBlock() during the columnar-to-row block boundary
materialisation, AFTER the scan has been planned with
oneTaskPerPartition=true (the DeltaScanRule-driven path that detects
input_file_name references and tells the task packer "one task per
partition so CometExecRDD's InputFileBlockHolder hook attributes every
row to the correct file").

Symptom in the Delta 4.1 full regression: many MERGE / UPDATE / DELETE
tests failed with DELTA_FILE_TO_OVERWRITE_NOT_FOUND because Delta's
getTouchedFile got input_file_name() == "" -- the absent thread-local
value -- and resolved that against dataPath, yielding the table-root
URI which is never in the AddFile map.

Local reproducer (new suite CometDeltaSpecialCharFilenameSuite):
  - "input_file_name returns real file path when target has MULTIPLE
    files (Delta MERGE shape)" -- the read-only failure the fix directly
    addresses. PASSES with the fix.
  - "CometDeltaNativeScanExec.oneTaskPerPartition is true when
    input_file_name is referenced" -- structural assertion on the
    executed plan. PASSES with the fix.
  - The two MERGE INTO reproducers are MARKED ignore: they still fail
    with the same DELTA_FILE_TO_OVERWRITE_NOT_FOUND shape after the fix
    because MERGE's plan-rewrite happens AFTER our DeltaScanRule runs,
    so the target-scan that findTouchedFiles joins on doesn't go through
    the same scan-with-oneTaskPerPartition rewrite. Tracked as a follow-up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Delta 4.1 own-test-suite regression failures on CDC reads under
row-tracking-enabled tables (especially the coordinated-commits batch
backfill variant DeltaCDCScalaWithCatalogOwnedBatch2Suite) surfaced as
'Comet Internal Error: Output column count mismatch: expected 14, got 13'.

The scan declared 14 output columns including the per-file
`default_row_commit_version` metadata column, but the native
DeltaSyntheticColumnsExec only knew about `base_row_id`, so the column
was dropped on the way through and the upstream operator saw N-1.

This commit:
  * Adds `default_row_commit_version` to JVM-side `syntheticNames` and
    `fixedMetadataNames` sets in CometDeltaNativeScan so it's included
    in `metadataColumnNamesEmitted` and the proto.
  * Adds the matching `META_DEFAULT_ROW_COMMIT_VERSION` constant, field
    schema, and emit branch in `synthetic_columns.rs`; extends
    `TaskMetadata` with the new field and wires it from
    `core_glue.rs` (the proto already carried the value via
    `task.default_row_commit_version`).

Removes the `DeltaSyntheticColumnsExec: unknown metadata column name
'default_row_commit_version'` failure path. A second off-by-one in the
same suite remains under investigation (separate column drop).

Also bundled:
  * `CometScanWithPlanData.perPartitionFilePaths` trait method +
    `operators.scala` union-path collector now matches the trait
    instead of just `CometNativeScanExec`, so MERGE/UPDATE/DELETE
    flows that embed a Delta scan in a parent native tree no longer
    see empty `input_file_name()` -> `DELTA_FILE_TO_OVERWRITE_NOT_FOUND`.
  * `CometExecRDD.compute` populates `InputFileBlockHolder` whenever
    `partition.filePaths.nonEmpty` (not only the single-file case),
    matching PR apache#3932's approach.
  * Re-enables the two MERGE reproducers in `CometDeltaSpecialCharFilenameSuite`.
  * Adds `DeltaHiveTest.scala` Comet-wiring hunk to `4.1.0.diff`
    (the piece PR apache#3932's 4.0.0.diff had that ours was missing).
  * New `CometDeltaRegressionReproSuite` (one repro per root-cause
    cluster identified in the 4.1 regression) and a `CometDeltaCdcSuite`.
  * New `.github/workflows/delta_regression_test.yml` workflow that
    invokes `contrib/delta/dev/run-regression.sh` across Delta 3.3.2 /
    4.0.0 / 4.1.0 with smoke -> full gating.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…etic schema augmentation

Follow-up to the metadata-column fix in 97c953a. The
`default_row_commit_version` column was added to `metadataColumnNamesEmitted`
and to the native synthetic-column emit branches, but
`isExtraSyntheticName` (which augments `requiredSchemaFields` with
metadata cols not in `scan.requiredSchema`) was missing it. The
resulting `requiredSchemaForProto` was one column short, so
`finalOutputIndices` was sized 13 instead of 14 and the native
`ProjectionExec` tried to read past the schema end.

Adds `default_row_commit_version` to the explicit name list alongside
`base_row_id`. Drops DeltaCDCScalaWithCatalogOwnedBatch2Suite from
6 -> 5 failures (the "CDC read's commit timestamps are correct
under different timezones" test now passes).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…d data length

When the Delta scan's `relation.dataSchema` includes a synthetic column
(e.g. `__delta_internal_is_row_deleted` from Delta's DV rewrite for
CDC reads of row-tracking-enabled tables), the JVM-side
`projectionVector` partition-tail indexes were computed off
`fileDataSchemaFields.length` (un-stripped, includes the synthetic),
but native receives `dataSchemaForProtoStripped` (synthetic removed
when `needsSyntheticEmit`). The partition tail then points one past
the end of the native data+partition schema, panicking
`ProjectionExprs::from_indices` with
"index out of bounds: the len is N but the index is N".

Compute partition-tail indexes from the non-synthetic data length
instead, mirroring what native sees. Also rewrite the
data-or-partition lookup in `requiredIndexes` to walk the
synthetic-filtered data field map so each entry points to the
correct native-side position.

Drops DeltaCDCScalaWithCatalogOwnedBatch2Suite from 5 -> 4 failures
("aggregating non-numeric cdc data columns" now passes). The
remaining 4 are "Results do not match" assertions (data correctness,
not native crashes) -- separate investigation.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CDC "delete events" / "insert events" reads attach a non-empty
`rowIndexFilters` map to `CdcAddFileIndex` / `TahoeRemoveFileIndex`.
This flips the DV bitmap meaning: native batch reads filter OUT the
rows in the bitmap, but CDC needs the rows IN the bitmap (the rows
being newly deleted/inserted). Our native scan only implements the
batch semantics, so it returned the wrong rows -- e.g. for a DELETE
"id > 20" on a file [20-24] with DV cardinality 4, the native scan
emitted the non-DV'd row (id=20) instead of the DV'd rows (21-24).

Add `DeltaReflection.hasInvertedRowIndexFilters` reflection helper
and gate `DeltaScanRule.nativeDeltaScan` to decline when set. Spark's
Delta reader then handles these correctly.

Result: DeltaCDCScalaWithCatalogOwnedBatch2Suite drops from 4 -> 0
failures (55/55 pass). Parent DeltaCDCScalaSuite still 55/55,
DeltaCDCSQLSuite still 54/54, contrib local Scala suites still
62 succeeded / 0 failed / 1 canceled (pre-existing).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ata-skipping

`refreshedSnapshotFiles` called `snapshot.filesForScan(Nil, false)` --
passing an empty filter list, so the refreshed file list was the FULL
table. This bypassed Delta's planning-time partition pruning and
stats-based data-skipping: for a PreparedDeltaFileIndex on a partitioned
table, our scan would read every file regardless of the predicate.

Surfaced as StatsCollectionSuite "gather stats" asserting
`recordsScanned(df.where("id = 1")) == 1` but Comet's native scan
returning 9 (all rows from all files).

Thread the scan's `partitionFilters ++ dataFilters` through
`extractBatchAddFiles` to `refreshedSnapshotFiles`, then to
`snapshot.filesForScan(filters, false)`. Delta evaluates partition
filters against partitionValues and data filters against AddFile
stats, returning only the surviving files.

Result: StatsCollectionSuite 65/67 -> 65 succeeded with "gather stats"
+ 1 other test now passing (was 4 failures, down to 2 -- the remaining
two read commit JSONs via LogicalRDD, separate code path).
DeltaCDCScalaWithCatalogOwnedBatch2Suite stays at 55/55; contrib local
suites stay at 62/62.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Mirrors MergeIntoSchemaEvolutionBaseExistingColumnSQLPathBasedCDCOnDVsPredPushOffSuite's
"upcast int source type into long target" failure shape: target written
with DV + CDC enabled, MERGE INSERT + UPDATE *, read back. Upstream
returns 0 rows (expected 4) because the post-MERGE read goes through
our scan applying a synthetic `__delta_internal_is_row_deleted` filter,
and something in the scan/Filter interaction drops all rows.

This local reproducer sets the equivalent session configs
(persistentDeletionVectors.enabled for merge, useMetadataRowIndex=false,
enableChangeDataFeed default, enableDeletionVectors default) and
performs the same operations, but does NOT yet fail. The Delta test
harness must set up additional state we're not mirroring. Committed
as-is so the repro is in place for the next iteration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Spark wraps file-source partition columns and per-batch constants in
`ConstantColumnVector`. Comet's `NativeUtil.exportBatch` and
`Utils.getBatchFieldVectors` previously threw
`Comet execution only takes Arrow Arrays, but got class
ConstantColumnVector` whenever such a batch reached a Comet operator --
notably OPTIMIZE on a Delta table with DVs, where the FileFormatWriter
threads a constant column through Comet's columnar pipeline.

Add `Utils.materializeConstantColumnVector` that allocates a fresh
Arrow `FieldVector` sized to the batch's row count and pre-fills it
with the constant value (or all-nulls when `isNullAt(0)`). Handles
the primitive types the OPTIMIZE / partition-column shapes surface:
Boolean, Byte, Short, Int, Long, Float, Double, Date, Timestamp
(MICROSECOND/UTC), String, Binary, Decimal128, Null. Throws for
unsupported types so we surface gaps loudly.

Wire it into both export paths so any Comet operator can ingest a
batch containing constants:
  - `NativeUtil.exportBatch`: materialise inline, then `Data.exportVector`
    through the existing C-Data path.
  - `Utils.getBatchFieldVectors` (used by `serializeBatches`):
    materialise into the field-vectors list ahead of the Arrow IPC
    write.

Result: OptimizeCompactionSQLSuite "optimize command with DVs" no
longer dies at NativeUtil.exportBatch with the ConstantColumnVector
error. The test now hits a separate failure (`NoSuchElementException`
during a later assertion that depends on our DV materialisation),
which is a distinct issue tracked separately.

No regressions: DeltaCDCScalaWithCatalogOwnedBatch2Suite 55/55,
contrib local Scala suites 63 succeeded / 0 failed / 1 canceled
(pre-existing), other 23 optimize tests still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…re read

`DeltaReflection.extractTableRoot` returns the double-URL-encoded URI
form (`pathToSingleEncodedUri` -> `path.toUri.toString`) so the NATIVE
side, which reads files via RawLocalFileSystem.pathToFile (which uses
the URI raw path verbatim), lands on the literal on-disk filename.

The JVM-side path was passing that same double-encoded string straight
into `new Path(tableRoot)` and on to `HadoopFileSystemDVStore.read`,
which resolves the DV file via Hadoop FS. Hadoop's raw path encoding
ended up doubly-encoded relative to the on-disk filename, so the DV
store opened a non-existent file and threw NPE on the null result.

`materializeDeletedRowIndexes` then returned `None`, our DeltaScanRule
treated that as a fatal DV-failure and declined, and the scan fell
back to Spark+Delta -- which on OPTIMIZE under DVs leads to a
NoSuchElementException downstream because a subsequent commit wasn't
written.

Fix: URL-decode `tableRoot` once before passing to `new Path(...)`.
Now Hadoop's URI form re-encodes back to the single-encoded form,
RawLocalFileSystem.pathToFile decodes once, and the DV store hits
the right on-disk file.

Result: `OptimizeCompactionSQLSuite "optimize command with DVs"`
now passes (1/1). No regressions: contrib local 63/63,
DeltaCDCScalaWithCatalogOwnedBatch2Suite still 55/55.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…=false

Delta's `deletionVectors.useMetadataRowIndex` is internal+default=true.
The DeletionVectorsTestUtils.PredicatePushdownDisabled mixin flips it
to false, switching Delta from parquet's `row_index` virtual column
to a synthetic `__delta_internal_is_row_deleted` column above the
scan.

In that mode, post-MERGE-with-persistentDV reads return wrong row
counts -- the original target files emit zero rows through our scan
path. The DV bitmap reads correctly, the synthetic emit logic mirrors
Delta's, but the batches downstream of the synthetic exec come back
empty (root cause is in the interaction between the Delta-side
filter rewrite + our per-file-group native plan + the DV synthetic
emit; needs deeper file-level investigation we can't land here).

Pragmatic fix: when `useMetadataRowIndex=false` AND
`__delta_internal_is_row_deleted` is in scan.requiredSchema AND any
AddFile carries a DV, decline. Spark+Delta then handles the read
correctly.

Production impact bounded: `useMetadataRowIndex` is internal+default
true, so production reads use parquet's `row_index` path (which our
scan handles correctly). The fallback only triggers for tests / users
who explicitly disable the conf.

Result: MergeIntoSchemaEvolutionBaseExistingColumnSQLPathBasedCDCOnDVsPredPushOffSuite
goes from 33/42 -> 42/42. No regression elsewhere:
DeltaCDCScalaWithCatalogOwnedBatch2Suite 55/55,
OptimizeCompactionSQLSuite 24/24, StatsCollectionSuite 65/67
(2 unrelated `recompute stats` failures), contrib local Scala 63/63.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…adata.file_path

When Spark packs multiple files into one Spark partition AND our scan
projects `_metadata.file_path` (the recompute-stats / per-file-aggregate
shape), we create multiple DataFusion file_groups (one per file) so
DeltaSyntheticColumnsExec emits a 1:1 per-file `file_path`
metadata mapping. But Spark consumes only ONE DataFusion partition per
Spark partition -- the 2nd+ files' batches are silently dropped, their
rows never reach Spark, and any per-file groupBy missing them.

Surfaced in StatsCollectionSuite "recompute stats multiple columns and
files" / "recompute stats on partitioned table": recompute does
`groupBy(_metadata.file_path).agg(min, max, ...)`. With 3 files and
2-file-per-partition packing, one file's rows go missing and its
recomputed stats come out null.

Fix: tag scans that project `file_path` with
`NeedsInputFileNameOption` (same plumbing as the
`input_file_name()` path), which propagates through to
`oneTaskPerPartition=true` on `CometDeltaNativeScanExec`. Each file
then becomes its own Spark partition with its own DataFusion
partition (1:1 alignment, no dropped data).

Narrow gate: only triggers when `file_path` is in scan.output. Other
per-file metadata cols (`base_row_id`, `default_row_commit_version`,
row-tracking helpers) commonly appear in unrelated scans and the
existing file packing handles them correctly.

Result: StatsCollectionSuite 65/67 -> 67/67. No regression:
OptimizeCompactionSQLSuite still 24/24, contrib local 64/64, parent
CDC 55/55 (separate suite still running, will verify).

Also drops a now-redundant local CometDeltaRegressionReproSuite stats
test that reproduces the path locally; it passes today because our
withDeltaTable doesn't trigger the multi-file packing that the
upstream test exercises, but the upstream test itself is the
authoritative reproducer.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Refactor `extract_storage_config` in the JNI bridge to delegate to a
pure `delta_storage_config_from_map`, making the kernel/Hadoop key
matrix testable in isolation. Removed the now-unused `map_get_string`
helper.

Add two test surfaces that document the known gaps so closing them
forces an obvious test update:

* `jni::tests::extract_storage_config_matrix` -- positive coverage
  for kernel-style keys, Hadoop-style fallbacks, and force-path-style.
* `jni::tests::extract_storage_config_known_gaps` -- negative
  assertions for GCS (`gcp_*`), per-bucket S3, and Hadoop Azure
  account/OAuth keys -- they pass today because the gap exists, and
  flip to failures the moment we wire those keys through.
* `CometDeltaCredentialAuditSuite` (Scala) -- pairs the native gaps
  with their JVM-side companions:
  - asserts s3/s3a/gs/wasb/wasbs prefixes extract the expected keys;
  - GAP: abfs/abfss schemes drop `fs.azure.*` keys because
    `NativeConfig.objectStoreConfigPrefixes` only registers
    `fs.abfs.` / `fs.abfss.` for them, missing OAuth/MSI creds that
    Hadoop users have historically set under `fs.azure.*`;
  - GAP: `augmentWithResolvedAwsCredentials` doesn't synthesize
    per-bucket `fs.s3a.bucket.<name>.*` keys after credential-chain
    resolution.

No behavior change for currently-supported credential paths; the
audit only adds regression coverage.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CometDeltaScanConfAuditSuite locks in the contract for every SQLConf /
table property that can change scan output. Each conf gets either a
CONTRACT_NATIVE assertion (engages + matches vanilla) or a
CONTRACT_FALLBACK assertion (declines + matches vanilla):

* DV: useMetadataRowIndex=false declines on DV-bearing reads
* DV: useMetadataRowIndex=true (default) engages native
* DV: MERGE_USE_PERSISTENT_DELETION_VECTORS=true read-side engages
* columnMapping.mode in {none,name,id} engages native
* rowTracking enabled does not disengage
* Comet kill switch (spark.comet.scan.deltaNative.enabled=false) forces
  fallback

Plus one GAP marker:
* path-based readChangeFeed routes through DeltaCDFRelation which our
  rule doesn't intercept -- table-API CDC reads (covered by
  CometDeltaCdcSuite) do engage. Flip this when the rule learns
  DeltaCDFRelation.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CometDeltaMetadataColumnAuditSuite projects every recognised
_metadata.* / row-tracking column individually and asserts the values
match vanilla Spark. Past bugs that motivated this:

* silent nulls / row drops when projecting _metadata.file_path on
  multi-file partitions (commit 8c3cf6c)
* column-count off-by-one for default_row_commit_version emit
  (commit 97c953a)
* synthetic-column stripped-length bug in projection_vector
  (commit 208f083)

Covers:
* _metadata.file_path / file_name / file_size / file_modification_time
* _metadata.file_block_start / file_block_length
* _metadata.row_index (row-tracking-enabled)
* input_file_name()
* _metadata.row_id (unmaterialised)
* _metadata.row_commit_version
* multi-metadata projection on RT + DV (the off-by-one repro shape)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CometDeltaTypeRoundTripAuditSuite covers Spark/Delta type round-trip
through the native scan vs vanilla:

Positive matrix:
* primitives at edge values (TINYINT/SMALLINT/INT/BIGINT extremes,
  FLOAT/DOUBLE +/-Infinity, NULLs)
* DECIMAL at (5,0) / (10,2) / (18,6) / (38,18)
* DATE / TIMESTAMP / TIMESTAMP_NTZ
* BINARY (empty, single byte, hex, long)
* nested STRUCT<>, ARRAY<>, MAP<>
* CHAR(N) / VARCHAR(N)

Discovered gaps (assert-and-document; promote when fixed):
* column mapping mode=name + complex types triggers AQE
  AdaptiveSparkPlanExec.setLogicalLinkForNewQueryStage assertion --
  likely missing/mismatched logical link on the serialized native
  scan node
* column mapping mode=id post-RENAME COLUMN triggers the same AQE
  assertion -- same root cause suspected
* VARIANT: probe + tolerant assertion (engages OR declines, but if it
  engages results must match vanilla -- guards against silent
  corruption when VARIANT support is added)

The CM-mode AQE finding is a follow-up worth its own ticket.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Delta stores partition values as strings in the log; coercion back to
the partition column's logical type happens at scan time. Past bugs
in this area are silent (wrong epoch, lost precision, double-applied
timezone), so we lock in the contract with native-vs-vanilla matches.

Matrix:
* DATE (epoch / current / max + partition-pruning filter)
* TIMESTAMP (UTC values, microsecond fractions)
* TIMESTAMP_NTZ (no TZ shift)
* DECIMAL(18,6) at sign/precision extremes
* BIGINT (Long.MinValue / MaxValue)
* STRING with spaces, percent-encoded literals, empty, NULL
* BOOLEAN
* Multi-column (dt, region) with IsNull pruning
* Session-TZ swap (write under UTC, read under LA) -- guards against
  double-applied timezone normalization

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CometDeltaFilterPushdownAuditSuite locks in correctness for the Spark
filter shapes that flow into the native scan (whether pushed down or
post-filtered, the result rows must match vanilla):

Positive matrix:
* EqualTo, EqualNullSafe (=, <=>)
* GreaterThan/LessThan/Ge/Le
* IsNull / IsNotNull on data column
* In / NotIn (incl. InSet-sized list)
* StringStartsWith / EndsWith / Contains
* AND/OR/NOT combinations
* Partition-column filters (EqualTo, In, GreaterThan)
* Nested struct field reference (s.a, s.b)
* Cast-coerced literal (INT literal vs BIGINT column)

Discovered gap (assert-and-document):
* DV-bearing table + range filter `id > 10 AND id < 25` -- native
  returns 19..24 while vanilla returns 11..24. Suspect stats-based
  data-skipping interaction with DV-bearing files reporting min/max
  that exclude DV'd rows. Filed as a follow-up; the GAP-marker test
  asserts native row count is LESS than vanilla and flips to a
  failure (with promotion guidance) once the mis-pruning is fixed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Audit 5 surfaced a silent correctness bug: a range filter over a
DV-bearing single-file table (rows 0..29, DELETE id%4=0) returned only
19..24 where vanilla Delta returns 11..24.

Root cause: when the synthetic `__delta_internal_is_row_deleted` column
is emitted, DeltaSyntheticColumnsStream walks a running
`current_row_offset` counter to look up each row's membership in the
file's `deleted_row_indexes` (the materialized DV). That counter assumes
the parquet reader returns EVERY row in physical order. But the same
scan also pushes `data_filters` down to the parquet reader, which skips
non-matching rows -- so `current_row_offset` no longer tracks the true
parquet row_index, and the DV bitmap gets applied to the wrong stream
positions (the first N rows of the post-filter stream rather than the
DV'd physical rows). The outer Spark Filter then drops those wrongly
flagged rows.

Fix: suppress data-filter pushdown to parquet when emit_is_row_deleted
is set (core_glue.rs). Spark's outer Filter still applies the predicates
correctly; we only forgo parquet-level pruning for this case. Partition
filters are unaffected (they prune file groups earlier). DV filtering
without the synthetic column (the common read path) still pushes filters
down normally.

Promotes the audit GAP-marker test in CometDeltaFilterPushdownAuditSuite
to a positive `assertDeltaNativeMatches`, plus two extra filters that
straddle the DV'd indexes. Verified no regression across the
features/CDC/metadata/scan-conf suites (28 tests).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Audit 3 surfaced an INTERNAL_ERROR on column-mapping tables: any plan
with a shuffle above the native Delta scan (orderBy / join / aggregate)
hit AdaptiveSparkPlanExec.setLogicalLinkForNewQueryStage's
`assert(link.isDefined)`. Originally thought to be complex-types or
post-rename specific; isolated to: CM-mode + ANY exchange.

Root cause: nativeDeltaScan propagated the logical link only from
`op.wrapped` (the wrapped FileSourceScanExec). In column-mapping mode
Delta builds that scan WITHOUT a logicalLink even though the surrounding
CometScanExec (`op`) has one. So the contrib's exec inherited no link,
and the built-in CometExecRule "set up logical links" pass -- which
re-derives a CometExec's link from `originalPlan.logicalLink` (==
`op.wrapped.logicalLink`) and UNSETS the tag when empty -- guaranteed the
exec ended up linkless. When AQE then wrapped it in a query stage at the
exchange boundary, the assertion fired.

Fix: seed the link as `op.wrapped.logicalLink.orElse(op.logicalLink)`,
set it on BOTH op.wrapped (so the downstream CometExecRule pass agrees)
and the exec. JVM-only change; no native rebuild required.

Promotes the two audit GAP markers in CometDeltaTypeRoundTripAuditSuite
to positive round-trip assertions and adds a minimal simple-types +
orderBy/aggregate regression that isolates the root cause. Verified no
regression across column-mapping / scan-conf / native / features suites
(34 tests).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…version

Delta own-suite regression (DeltaTimeTravelSuite / DeltaHistoryManager*):
a versionAsOf/timestampAsOf read returned the LATEST version's data
(e.g. v0 returned 20 rows instead of 10).

Root cause: DeltaReflection.refreshedSnapshotFiles (used by
extractBatchAddFiles for PreparedDeltaFileIndex) unconditionally called
deltaLog.update() to refresh the snapshot to HEAD before filesForScan.
That refresh exists to pick up fresh deletion-vector descriptors for
consecutive DELETEs that reuse a cached FileIndex -- but for a
time-travel query it discards the pinned version and returns current
files.

Fix: gate the head-refresh on PreparedDeltaFileIndex.versionScanned.
When it's Some(v) (time travel), use the pinned preparedScan.scannedSnapshot
instead -- historical versions are immutable so the DV-staleness reason
to refresh does not apply. Non-time-travel reads keep the head refresh.

Reproduced by CometDeltaTimeTravelReproSuite (versionAsOf=0 must return
v0's rows, not head); fails before / passes after. JVM-only change.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…TT gate

Follow-up to 7c59938. Rather than gate the head-refresh on
versionScanned (a band-aid), drop the head-refresh entirely: always
re-query `preparedScan.scannedSnapshot` -- the snapshot the scan was
prepared against, which is exactly what vanilla Spark+Delta reads
(PreparedDeltaFileIndex extends TahoeFileIndexWithSnapshotDescriptor
over it).

The earlier `deltaLog.update()` to head (commit 830c979) was meant to
pick up DV descriptors written after a cached FileIndex was built, but
refreshing to head makes Comet read a DIFFERENT snapshot than vanilla:
it diverges on the consecutive-DELETE / DeltaLog-cache case (that
commit's own point apache#2 acknowledges vanilla returns the "stale" count)
and it returned the LATEST version for time-travel reads. Re-querying
the prepared snapshot via filesForScan still picks up that snapshot's
freshest DV descriptors, matches vanilla in every case, and needs no
time-travel special-casing.

Renamed refreshedSnapshotFiles -> preparedSnapshotFiles. Verified: 24
contrib tests pass incl. the DV double-DELETE (CometDeltaColumnMapping
Suite) the head-refresh was added for, and CometDeltaTimeTravelRepro
Suite. JVM-only.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>